package vn.com.epi.utils;

import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import vn.com.epi.data.table.ArticleTable;
import vn.com.epi.data.table.UserTable;
import vn.com.epi.log.Import.LogConcatenate;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

public class LogCleaner extends Configured implements Tool {
    
    private final Log log = LogFactory.getLog(LogCleaner.class);

    public static final long ONE_DAY = 86400; 
    public static final long TIME_LIMIT = ONE_DAY * 5; // 5 days
    public static final int GROUP_SIZE = 1000;
    
    private Map<String, Integer> map;
    
    public void deleteOldVisitedArticles() throws IOException {
        log.info("----------------* start delete old visited article *------------------");
        HTable table = new HTable(UserTable.NAME);
        map = Maps.newHashMap();
        try {
            List<Put> puts = Lists.newArrayList();
            int counter = 0;
            Scan scan = new Scan();
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES);        
            scan.addColumn(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS);        
            ResultScanner rs = table.getScanner(scan);
            long current = System.currentTimeMillis() / 1000L;
            for (Result r: rs) {
                long maxTimestamp = 0;
                int totalVisitPerDay = 0;
                counter++;
                StringBuilder articleSb = new StringBuilder();
                StringBuilder timestampSb = new StringBuilder();
                String timestampCol = Bytes.toString(r.getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS));
                if (!Strings.isNullOrEmpty(timestampCol)) {
                    String articleCol = Bytes.toString(r.getValue(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES));
                    String timestamps[] = timestampCol.split(",");
                    String[] articles = articleCol.split(",");
                    for (int i = 0; i < articles.length; i++) {
                        long timestamp = Long.parseLong(timestamps[i]);
                        maxTimestamp = Math.max(maxTimestamp, timestamp);
                        if (current - timestamp < TIME_LIMIT) {                            
                            articleSb.append(articles[i]).append(",");
                            timestampSb.append(timestamps[i]).append(",");
                        }
                        if (current - timestamp < ONE_DAY) {
                        	totalVisitPerDay++;
                        }

                    }
                    StringUtils.deleteLastChar(articleSb);
                    StringUtils.deleteLastChar(timestampSb);
                }
                String userId = Bytes.toString(r.getRow());
                map.put(userId, totalVisitPerDay);
                Put put = new Put(r.getRow());
                put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_ARTICLES, Bytes.toBytes(articleSb.toString()));
                put.add(UserTable.FAM_MAIN, UserTable.COL_VISITED_TIMESTAMPS, Bytes.toBytes(timestampSb.toString()));
                puts.add(put);
                if (counter == GROUP_SIZE) {                    
                    counter = 0;
                    table.put(puts);
                    puts = Lists.newArrayList();    
                }
            }
            if (counter > 0)
                table.put(puts);
            log.info("----------------* completed delete articles *------------------");            
        } catch (Exception e) {
          // ignore
            System.out.println("Error while delete old articles: " + e.toString());
            log.info("Error while delete old articles !!!!");
        } finally {
            table.close();
        } 
    }
    
    public void deleteNoiseUsers() throws IOException {
        log.info("----------------* start delete noise users *------------------");
        HTable userTable = new HTable(UserTable.NAME);
        int deletedUsers = 0;
        try {
            List<Delete> deletes = Lists.newArrayList();
            int counter = 0;
            for (String userId: map.keySet()) {
            	int totalVisited = map.get(userId); 
                if (totalVisited > 100) {
                	byte[] row = Bytes.toBytes(userId);
                    Delete delete = new Delete(row);
					deletes.add(delete);
					counter++;
					deletedUsers++;
                }                        
                if (counter == GROUP_SIZE) {
                    userTable.delete(deletes);
                    deletes = Lists.newArrayList();
                    counter = 0;
                }
            }
            if (counter > 0) {
                userTable.delete(deletes);
            }
            log.info("----------------* completed delete users *------------------");
            System.out.println("Deleted " + deletedUsers + " users");
        } catch (Exception e) {
          // ignore
            System.out.println("Error while delete users: " + e.toString());
            log.info("Error while delete users !!!!");
        } finally {
        	userTable.close();
        }         
    }
    
    public void deleteOldArticles() throws IOException {
        log.info("----------------* start delete old article *------------------");
        HTable articleTable = new HTable(ArticleTable.NAME);
        int deletedArticles = 0;
        try {
            List<Delete> deletes = Lists.newArrayList();
            Scan scan = new Scan();
            scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS);        
            ResultScanner rs = articleTable.getScanner(scan);
            long current = System.currentTimeMillis() / 1000L;
            int counter = 0;
            for (Result r: rs) {
                long firstVisitedTimestamp = Bytes.toLong(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_FIRST_VISIT_TIMESTAMPS));
                if (current - firstVisitedTimestamp > TIME_LIMIT) {
                    Delete delete = new Delete(r.getRow());
                    deletes.add(delete);
                    counter++;
                    deletedArticles++;
                }
                if (counter == GROUP_SIZE) {
                    articleTable.delete(deletes);
                    deletes = Lists.newArrayList();
                    counter = 0;
                }
            }
            if (counter > 0) {
                articleTable.delete(deletes);
            }
            log.info("----------------* completed delete articles *------------------");            
            System.out.println("Deleted " + deletedArticles + " articles");
        } catch (Exception e) {
          // ignore
            System.out.println("Error while delete article: " + e.toString());
            log.info("Error while delete article !!!!");
        } finally {
            articleTable.close();
        }         
    }    

    @Override
    public int run(String[] args) throws Exception {
    	LogConcatenate.concat();
        deleteOldVisitedArticles();
        deleteNoiseUsers();
        deleteOldArticles();        
        return 0;
    }
    
    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ToolRunner.run(new LogCleaner(), args);
        long endTime = System.currentTimeMillis();
        System.out.println("total time used: " + (endTime - startTime) + " ms.");        
    }
    
}